package club.monkeywood.ad.dmp.report

import java.util.Properties

import com.typesafe.config.ConfigFactory
import org.apache.spark.SparkConf
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

/**
	* 广告投放的地域分布统计
	* 实现方式：
	* Spark SQL
	*/
object AreaAnalyseRptBySql {

	def main(args: Array[String]): Unit = {

		// 0 校验参数个数
		if (args.length != 1) {
			println(
				"""
					|club.monkeywood.ad.dmp.report.AreaAnalyseRptBySql
					|参数：
					| logInputPath
				""".stripMargin)
			sys.exit()
		}

		// 1 接受程序参数
		val Array(logInputPath) = args

		// 2 创建sparkconf->SparkSession
		val sparkConf = new SparkConf()
		sparkConf.setAppName(s"${this.getClass.getSimpleName}")
		sparkConf.setMaster("local[*]")
		//使用KryoSerializer更快
		// RDD 序列化到磁盘 worker与worker之间的数据传输
		sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

		val ss = SparkSession.builder()
			.appName(s"${this.getClass.getSimpleName}")
			.config(sparkConf)
			.getOrCreate()

		// 读取数据 --》parquet文件
		val df: DataFrame = ss.read.parquet(logInputPath)

		// 将dataframe注册成一张临时表
		df.createTempView("log")

		// 按照省市进行分组聚合---》统计分组后的各省市的日志记录条数
		// 业务逻辑
		val result: DataFrame = ss.sql(
			"""
				|select
				|provincename, cityname,
				|sum(case when requestmode=1 and processnode >=2 then 1 else 0 end) valid_request,
				|sum(case when requestmode=1 and processnode =3 then 1 else 0 end) ad_request,
				|sum(case when iseffective=1 and isbilling=1 and isbid=1 and adorderid !=0 then 1 else 0 end) compete_num,
				|sum(case when iseffective=1 and isbilling=1 and iswin=1 then 1 else 0 end) compete_success_num,
				|sum(case when requestmode=2 and iseffective=1 then 1 else 0 end) show_num,
				|sum(case when requestmode=3 and iseffective=1 then 1 else 0 end) click_num,
				|sum(case when iseffective=1 and isbilling=1 and iswin=1 then 1.0*adpayment/1000 else 0 end) ad_cost,
				|sum(case when iseffective=1 and isbilling=1 and iswin=1 then 1.0*winprice/1000 else 0 end) ad_consume
				|from log
				|group by provincename, cityname
			""".stripMargin)

		result.show()

		// 自动从资源文件中加载配置文件
		// 查找顺序：application.conf -> application.json --> application.properties
		val load = ConfigFactory.load()
		val url = load.getString("jdbc.url")
		val tableName = load.getString("jdbc.tableName")
		val props = new Properties()
		props.setProperty("user", load.getString("jdbc.user"))
		props.setProperty("password", load.getString("jdbc.password"))

		// 将结果写入到mysql的 rpt_pc_count 表中
		result.write
			//插入新数据
			//默认创建新表
			.mode(SaveMode.Append)
			.jdbc(url, tableName, props)

		ss.stop()
	}

}
